package com.katesoft.scale4j.rttp.jobs.jdbc.init;

import static com.katesoft.scale4j.rttp.internal.IRttpHazelcastRefs.SCHEDULER_QUARTZ_DB_CREATION_LOCK;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;
import java.util.LinkedHashSet;

import net.jcip.annotations.ThreadSafe;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.perf4j.StopWatch;
import org.springframework.core.io.Resource;
import org.springframework.jdbc.datasource.init.ScriptStatementFailedException;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ILock;

/**
 * This class will create quartz schema during startup if necessary.
 * <p/>
 * This class is designed to be used within jvmcluster. first will obtain distributed lock and then
 * create schema if necessary.
 * 
 * @author kate2007
 */
@ThreadSafe
public class RttpQuartzSchemaCreator extends AbstractSchemaCreator {
   public RttpQuartzSchemaCreator() {
      tablesPrefix = "QRTZ_";
   }

   /**
    * will create scheduler schema if does not exists
    * 
    * @param connection
    *           database connection
    * @throws SQLException
    *            re-throw any SQLExceptions
    */
   @Override
   public void populate(Connection connection) throws SQLException {
      if (getSchedulerCreationScript() != null) {
         StopWatch stopWatch = new StopWatch(SCHEDULER_QUARTZ_DB_CREATION_LOCK);
         HazelcastInstance instance = springHazelcastBridge.getRunningInstance();

         ILock lock = instance.getLock(SCHEDULER_QUARTZ_DB_CREATION_LOCK);
         logger.info("using %s and dialect %s for scheduler schema creation", lock.toString(),
                  dialect.toString());
         try {
            if (lock.tryLock()) {
               try {
                  if (!tablesCreated(connection)) {
                     if (!springHazelcastBridge.getAtomicBoolean(SCHEDULER_QUARTZ_DB_CREATION_LOCK)
                              .get()) {
                        logger.info(
                                 "scheduler schema is not created, will create using connection %s",
                                 connection);
                        springHazelcastBridge.updateAtomicBoolean(
                                 SCHEDULER_QUARTZ_DB_CREATION_LOCK, true);
                        super.setScripts(new Resource[] { getSchedulerCreationScript() });
                        try {
                           super.populate(connection);
                        } catch (ScriptStatementFailedException e) {
                           logger.warn(
                                    "unable to create quartz schema for dialect %s, exception = %s",
                                    dialect.getSimpleName(), ExceptionUtils.getFullStackTrace(e));
                        }
                     }
                  }
               } finally {
                  lock.unlock();
               }
            }
         } finally {
            logger.debug("scheduler schema created in %s milisecs using %s",
                     stopWatch.getElapsedTime(), connection);
         }
      }
   }

   @Override
   public Collection<String> tables() {
      Collection<String> tables = new LinkedHashSet<String>();

      tables.add(tablesPrefix + "JOB_DETAILS");
      tables.add(tablesPrefix + "JOB_LISTENERS");
      tables.add(tablesPrefix + "TRIGGERS");
      tables.add(tablesPrefix + "SIMPLE_TRIGGERS");
      tables.add(tablesPrefix + "CRON_TRIGGERS");
      tables.add(tablesPrefix + "BLOB_TRIGGERS");
      tables.add(tablesPrefix + "TRIGGER_LISTENERS");
      tables.add(tablesPrefix + "CALENDARS");
      tables.add(tablesPrefix + "PAUSED_TRIGGER_GRPS");
      tables.add(tablesPrefix + "FIRED_TRIGGERS");
      tables.add(tablesPrefix + "SCHEDULER_STATE");
      tables.add(tablesPrefix + "LOCKS");

      return tables;
   }
}
